package com.example.yckjbigdataservice.controller;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.sql.*;
import java.util.Properties;

@RequestMapping("/phoenix")
@RestController
@Slf4j
public class PhoenixController {

    private final Properties phoenixProperties;

    public PhoenixController(@Qualifier("phoenixProperties") Properties phoenixProperties) {
        this.phoenixProperties = phoenixProperties;
    }

    @GetMapping("/create/table")
    public String createTable() {
        testCreateTable();
        return "创建默认的表success";
    }

    @GetMapping("/create/table/mapping")
    public String createTableMapping() {

        /**
        Hbase未开启phoenix
        */
        String url = "jdbc:phoenix:" + phoenixProperties.get("hbase.zookeeper.quorum");

        try (Connection conn = DriverManager.getConnection(url, phoenixProperties);
             Statement stmt = conn.createStatement()) {

            // 对于族列名不是cf的，叫f1的，查询也要加f1.name
            /*String ddl = "CREATE TABLE IF NOT EXISTS my_table ("
                    + "id VARCHAR PRIMARY KEY, "
                    + "\"f1\".\"name\" VARCHAR, "
                    + "\"f1\".\"age\" INTEGER"
                    + ") COLUMN_ENCODED_BYTES=0 "
                    + "STORED BY 'org.apache.hadoop.hbase.regionserver.IndexedKeyValueStore' "
                    + "WITH ('hbase.table.name'='my_ns:my_table')";*/

            String ddl = "CREATE TABLE IF NOT EXISTS std_16LZDR ("
                    + "id VARCHAR PRIMARY KEY, "
                    + "CQ00027_DE00001 VARCHAR, "
                    + "CQ00027_DE00002 VARCHAR"
                    + ") COLUMN_ENCODED_BYTES=0 "
                    + "STORED BY 'org.apache.hadoop.hbase.regionserver.IndexedKeyValueStore' "
                    + "WITH ('hbase.table.name'='std:std_16LZDR')";


            // 创建视图 才能对字段进行别名，创建映射必须要一致
            /*CREATE VIEW my_table_view (
                    username,    -- 对应 "f1"."name"
                    user_age     -- 对应 "f1"."age"
            )
            AS SELECT id, "f1"."name", "f1"."age" FROM my_table;*/

            stmt.execute(ddl);
            System.out.println("Phoenix 映射建立成功！");
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
        return "Phoenix 映射建立成功！";
    }

    @GetMapping("/data/select")
    public String scanData(@RequestParam("tableName") String tableName,
                           @RequestParam("field") String field,
                           @RequestParam("value") String value) {
        log.info("phoenix开始查询,hbase.zookeeper.quorum={}",phoenixProperties.get("hbase.zookeeper.quorum"));

        String url = "jdbc:phoenix:" + phoenixProperties.get("hbase.zookeeper.quorum");
        //String url = "jdbc:phoenix:" + phoenixProperties.get("hbase.zookeeper.quorum") + "/hbase";
        // Query
        String querySQL = "SELECT * FROM "+ tableName + " WHERE " + field +" = ?";
        java.sql.Connection conn = null;
        PreparedStatement preStat = null;
        Statement stat = null;
        ResultSet result = null;
        try {
            // Create Connection
            conn = DriverManager.getConnection(url, phoenixProperties);
            // Create Statement
            stat = conn.createStatement();
            // Create PrepareStatement
            preStat = conn.prepareStatement(querySQL);
            // Execute query
            preStat.setString(1, value);
            result = preStat.executeQuery();
            // Get result
            while (result.next()) {
                int columnCount = result.getMetaData().getColumnCount();
                for (int i = 1; i <= columnCount; i++) {
                    String columnName = result.getMetaData().getColumnName(i);
                    log.info("{} = {}", columnName, result.getObject(columnName).toString());
                }
                /*int id = result.getInt("id");
                String name = result.getString(1);
                log.info("id: {} name: {}", id, name);*/
            }
            log.info("Select successfully.");
        } catch (SQLException e) {
            log.error("Select failed.", e);
        } finally {
            if (null != result) {
                try {
                    result.close();
                } catch (SQLException e2) {
                    log.error("Result close failed.", e2);
                }
            }
            if (null != stat) {
                try {
                    stat.close();
                } catch (SQLException e2) {
                    log.error("Stat close failed.", e2);
                }
            }
            if (null != conn) {
                try {
                    conn.close();
                } catch (SQLException e2) {
                    log.error("Connection close failed.", e2);
                }
            }
        }
        return "phoenix数据查询success";
    }
    /**
     * PhoenixSample test
     */
    public void test() {
        testCreateTable();
        testPut();
        testSelect();
        testDrop();
    }

    /**
     * Create Table
     */
    public void testCreateTable() {
        log.info("Entering testCreateTable.");
        String url = "jdbc:phoenix:" + phoenixProperties.get("hbase.zookeeper.quorum");
        // Create table
        String createTableSQL =
                "CREATE TABLE IF NOT EXISTS TEST (id integer not null primary key, name varchar, "
                        + "account char(6), birth date)";
        try (java.sql.Connection conn = DriverManager.getConnection(url, phoenixProperties);
             Statement stat = conn.createStatement()) {
            // Execute Create SQL
            stat.executeUpdate(createTableSQL);
            log.info("Create table successfully.");
        } catch (SQLException e) {
            log.error("Create table failed.", e);
        }
        log.info("Exiting testCreateTable.");
    }

    /**
     * Put data
     */
    public void testPut() {
        log.info("Entering testPut.");
        String url = "jdbc:phoenix:" + phoenixProperties.get("hbase.zookeeper.quorum");
        // Insert
        String upsertSQL =
                "UPSERT INTO TEST VALUES(1,'John','100000', TO_DATE('1980-01-01','yyyy-MM-dd'))";
        try (java.sql.Connection conn = DriverManager.getConnection(url, phoenixProperties);
             Statement stat = conn.createStatement()) {
            // Execute Update SQL
            stat.executeUpdate(upsertSQL);
            conn.commit();
            log.info("Put successfully.");
        } catch (SQLException e) {
            log.error("Put failed.", e);
        }
        log.info("Exiting testPut.");
    }

    /**
     * Select Data
     */
    public void testSelect() {
        log.info("Entering testSelect.");
        String url = "jdbc:phoenix:" + phoenixProperties.get("hbase.zookeeper.quorum");
        // Query
        String querySQL = "SELECT * FROM TEST WHERE id = ?";
        java.sql.Connection conn = null;
        PreparedStatement preStat = null;
        Statement stat = null;
        ResultSet result = null;
        try {
            // Create Connection
            conn = DriverManager.getConnection(url, phoenixProperties);
            // Create Statement
            stat = conn.createStatement();
            // Create PrepareStatement
            preStat = conn.prepareStatement(querySQL);
            // Execute query
            preStat.setInt(1, 1);
            result = preStat.executeQuery();
            // Get result
            while (result.next()) {
                int id = result.getInt("id");
                String name = result.getString(1);
                log.info("id: {} name: {}", id, name);
            }
            log.info("Select successfully.");
        } catch (SQLException e) {
            log.error("Select failed.", e);
        } finally {
            if (null != result) {
                try {
                    result.close();
                } catch (SQLException e2) {
                    log.error("Result close failed.", e2);
                }
            }
            if (null != stat) {
                try {
                    stat.close();
                } catch (SQLException e2) {
                    log.error("Stat close failed.", e2);
                }
            }
            if (null != conn) {
                try {
                    conn.close();
                } catch (SQLException e2) {
                    log.error("Connection close failed.", e2);
                }
            }
        }
        log.info("Exiting testSelect.");
    }

    /**
     * Drop Table
     */
    public void testDrop() {
        log.info("Entering testDrop.");
        String url = "jdbc:phoenix:" + phoenixProperties.get("hbase.zookeeper.quorum");
        // Delete table
        String dropTableSQL = "DROP TABLE TEST";

        try (Connection conn = DriverManager.getConnection(url, phoenixProperties);
             Statement stat = conn.createStatement()) {
            stat.executeUpdate(dropTableSQL);
            log.info("Drop successfully.");
        } catch (SQLException e) {
            log.error("Drop failed.", e);
        }
        log.info("Exiting testDrop.");
    }
}
